package com.lagou.bak;

import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.*;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import javax.annotation.Nonnull;
import java.io.IOException;
import java.util.Collection;
import java.util.Properties;

public class KafkaCheckPointDemo {
    static int count = 0;
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(10);
//        env.getCheckpointConfig().setCheckpointInterval(10);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        env.setStateBackend(new FsStateBackend("file:////tmp/backend"));

//        String topic = "animalone";
        String topic = "animalone";
        Properties props = new Properties();
//        props.setProperty("bootstrap.servers", "hadoop2:9092");
        props.setProperty("bootstrap.servers", "hdp-2:9092");
        props.setProperty("group.id", "mygp");

        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), props);
        consumer.setStartFromEarliest();
        consumer.setCommitOffsetsOnCheckpoints(true);
        DataStreamSource<String> messageStream = env.addSource(consumer);

        SingleOutputStreamOperator<String> maped = messageStream.rebalance().map(new MapFunction<String, String>() {

            public String map(String s) throws Exception {
                if (count++ == 20000) {
//                    Thread.sleep(1000);
                    count = 0;
                    throw new Exception("Failed.");
                }
                if(Integer.valueOf(s)%100 == 0) {
                    System.out.println(s);
                }

                return s;
            }
        });

//        maped.print();

        env.execute();
    }

}
